use std::{
    collections::HashMap,
    mem::MaybeUninit,
    sync::{atomic::AtomicBool, Arc, Mutex},
    time::Duration,
};

use bytes::Bytes;
use common::{service::sys_service_api::http::{self, HTTPMethod}, base::status::Status};
use hyper::{
    body::HttpBody,
    http::{HeaderName, HeaderValue},
    server::conn::Http,
    service::service_fn,
    Body, Client, Method, Request, Response,
};
use hyper_tls::HttpsConnector;
use tokio::{net::TcpListener, sync::oneshot::Sender, time::sleep};
use tracing::{error, info, warn, debug};

pub struct HttpClient {
    pub(super) request_id: i64,
    pub res: MaybeUninit<Response<Body>>,
}
impl HttpClient {
    pub fn new(request_id: i64) -> HttpClient {
        let http_client = HttpClient {
            request_id,
            res: MaybeUninit::uninit(),
        };
        http_client
    }

    pub async fn do_request(&mut self, request: http::HttpRequest) -> Result<(), Status> {
        let method = if let Some(method) = request.method {
            match method {
                HTTPMethod::GET => Method::GET,
                HTTPMethod::POST => Method::POST,
            }
        } else {
            Method::GET
        };
        let body = request.body;

        let mut req = Request::builder().method(method).uri(&request.url);

        let headers = req.headers_mut().unwrap();

        for (key, value) in request.headers {
            headers.insert(
                HeaderName::from_bytes(key.as_bytes()).unwrap(),
                HeaderValue::from_str(&value).unwrap(),
            );
        }

        debug!("body = {}", body);

        let req = req.body(Body::from(body)).unwrap();

        let res = if req.uri().scheme_str() == Some("https") {
            let https = HttpsConnector::new();
            let client = Client::builder().build(https);
            client.request(req).await
        } else {
            let client = Client::new();
            client.request(req).await
        };

        if res.is_err() {
            return Err(Status::error(res.err().unwrap().to_string()));
        }
        let res = res.unwrap();
        unsafe {
            self.res.as_mut_ptr().write(res);
        }
        Ok(())
    }

    pub async fn get_data(&mut self) -> Result<Bytes, Status> {
        unsafe {
            if self.res.assume_init_mut().is_end_stream() {
                return Ok(Bytes::new());
            }
        }

        let data = unsafe { self.res.assume_init_mut().data().await };

        if data.is_none() {
            return Err(Status::error("读取数据失败！".into()));
        }

        let data = data.unwrap();
        if data.is_err() {
            return Err(Status::error("读取数据失败！".into()));
        }

        Ok(data.unwrap())
    }
}

mod router {
    use common::protocol::protocol_v1::ProtocolV1Reader;
    use common::serial::request_message::RequestMessage;
    use common::{
        base::{id_generator, status::Status},
        errors::serial_errors::SerialError,
        protocol::{protocol_v1::ProtocolV1Writer, MsgType, XID},
        serial::{
            api_descriptor::ServiceApiDescriptor,
            json_to_proto::{compute_size, json_to_proto_with_buffer},
            proto_to_json::proto_to_json_by_is,
        },
        service::ServiceKey,
    };
    use hyper::{body::HttpBody, Body, Method, Request, Response, StatusCode};

    use protobuf::{descriptor::MethodDescriptorProto, CodedInputStream, CodedOutputStream};
    use regex::{Regex, RegexBuilder};
    use serde::{Deserialize, Serialize};
    use serde_json::Value;

    use std::{convert::Infallible, mem::MaybeUninit};
    use tokio::{
        fs::File,
        io::{AsyncWriteExt, BufWriter},
        sync::oneshot,
    };
    use tokio_util::codec::{BytesCodec, FramedRead};
    use tracing::{info, warn};

    use crate::application::Application;

    #[derive(Serialize, Deserialize)]
    struct InvokeRequest {
        pub(super) dxc_name: String,
        pub(super) dxc_version: String,
        pub(super) service_name: String,
        pub(super) api: String,
        pub(super) params: Value,
    }

    #[derive(Serialize, Deserialize)]
    struct ResponseMessage {
        err_code: i32,
        err_msg: String,
        data: Option<Value>,
    }

    static mut DOWNLOAD_REGX: MaybeUninit<Regex> = MaybeUninit::uninit();

    static mut UPLOAD_REGX: MaybeUninit<Regex> = MaybeUninit::uninit();

    fn build_route_regex(path: &str) -> Regex {
        let regex_pattern = path.replace("{", "(?P<").replace("}", ">[^/]+)");
        let regex_pattern = format!("^{}$", regex_pattern);

        RegexBuilder::new(&regex_pattern)
            .case_insensitive(true)
            .build()
            .unwrap()
    }

    pub fn init() {
        let download_regx = build_route_regex("/download/{key}");
        let upload_regx = build_route_regex("/upload/{key}");

        unsafe {
            DOWNLOAD_REGX.as_mut_ptr().write(download_regx);
            UPLOAD_REGX.as_mut_ptr().write(upload_regx);
        }
    }

    fn new_http_service_key() -> ServiceKey {
        ServiceKey::new("HttpService".into(), "0.0.1".into(), "HttpService".into())
    }
    /**
     *
     */
    fn new_receiver_service_key(invoke_request: InvokeRequest) -> ServiceKey {
        ServiceKey::new(
            invoke_request.dxc_name,
            invoke_request.dxc_version,
            invoke_request.service_name,
        )
    }

    async fn handle_download(
        _req: Request<Body>,
        _key: String,
    ) -> Result<Response<Body>, Infallible> {
        let file_path = "./demo.txt";

        if let Ok(file) = File::open(file_path).await {
            let stream = FramedRead::new(file, BytesCodec::new());
            let body = Body::wrap_stream(stream);
            return Ok(Response::new(body));
        }
        //
        Ok(Response::builder()
            .status(StatusCode::NOT_FOUND)
            .body(Body::from("文件不存在！"))
            .unwrap())
    }

    async fn handle_upload(
        _req: Request<Body>,
        _key: String,
    ) -> Result<Response<Body>, Infallible> {
        let temp_file_path = "./temp.bin";

        let temp_file = File::create(temp_file_path).await;

        if temp_file.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .body(Body::from("创建文件失败，请稍后再试！"))
                .unwrap());
        }
        let temp_file = temp_file.unwrap();
        let mut writer = BufWriter::new(temp_file);
        let mut body = _req.into_body();
        let mut is_err = false;
        while let Some(bytes) = body.data().await {
            if bytes.is_err() {
                is_err = true;
                break;
            }
            let mut bytes = bytes.unwrap();

            let ret = writer.write_buf(&mut bytes).await;

            if ret.is_err() {
                is_err = true;
                break;
            }
        }
        if is_err {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from("创建文件失败，请稍后再试！"))
                .unwrap());
        }

        let ret = writer.flush().await;

        if ret.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from("创建文件失败，请稍后再试！"))
                .unwrap());
        }

        Ok(Response::builder()
            .status(StatusCode::OK)
            .body(Body::empty())
            .unwrap())
    }

    fn request_to_proto(
        method: &MethodDescriptorProto,
        invoke_request: &InvokeRequest,
        service_api_descriptor: &ServiceApiDescriptor,
    ) -> Result<Vec<u8>, SerialError> {
        let has_input_type =
            method.has_input_type() && method.input_type() != ".google.protobuf.Empty";

        let mut buffer_size = protobuf::rt::string_size(1, method.name());
        // 接收者
        if has_input_type {
            buffer_size += compute_size(
                &invoke_request.params,
                method.input_type(),
                service_api_descriptor,
            )?;
        }
        let mut buffer = Vec::with_capacity(buffer_size as usize);
        let mut output_stream = CodedOutputStream::vec(&mut buffer);
        output_stream.write_string(1, method.name()).unwrap();

        if has_input_type {
            json_to_proto_with_buffer(
                &invoke_request.params,
                method.input_type(),
                service_api_descriptor,
                &mut output_stream,
            )?;
        }

        drop(output_stream);

        Ok(buffer)
    }

    fn response_proto_to_json_value(
        method: &MethodDescriptorProto,
        service_api_descriptor: &ServiceApiDescriptor,
        is: &mut CodedInputStream,
    ) -> Option<Value> {
        if !method.has_output_type() && method.output_type() != ".google.protobuf.Empty" {
            return None;
        }
        let result = proto_to_json_by_is(method.output_type(), service_api_descriptor, is);

        return if let Ok(value) = result {
            Some(value)
        } else {
            None
        };
    }

    async fn handle_invoke(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
        let whole_body = hyper::body::to_bytes(_req.into_body()).await;

        if whole_body.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::empty())
                .unwrap());
        }

        let whole_body = whole_body.unwrap();

        let whole_body = whole_body.as_ref();

        let invoke_request = serde_json::from_slice::<InvokeRequest>(whole_body);

        if invoke_request.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from(invoke_request.err().unwrap().to_string()))
                .unwrap());
        }
        let invoke_request = invoke_request.unwrap();
        let airport = Application::get_airport();
        let dxc_version = semver::Version::parse(&invoke_request.dxc_version);
        if dxc_version.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from("版本号格式错误！"))
                .unwrap());
        }
        let dxc_version = dxc_version.unwrap();

        let dxc = airport
            .get_dxc_manager()
            .get_dxc_by_name_and_version_without_build(&invoke_request.dxc_name, &dxc_version)
            .await;

        if dxc.is_none() {
            return Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Body::from("服务不存在！"))
                .unwrap());
        }
        //
        let service = dxc
            .unwrap()
            .get_local_service_by_name(&invoke_request.service_name)
            .await;

        if service.is_none() {
            return Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Body::from("服务不存在！"))
                .unwrap());
        }
        // 对象存在，消息存在
        let service = service.unwrap();
        let method = service.api_descriptor.method(&invoke_request.api);
        if method.is_none() {
            return Ok(Response::builder()
                .status(StatusCode::NOT_FOUND)
                .body(Body::from("接口不存在！"))
                .unwrap());
        }
        let method = method.unwrap();
        let message = request_to_proto(method, &invoke_request, &service.api_descriptor);
        if message.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body(Body::from(message.err().unwrap().message))
                .unwrap());
        }

        let message: Vec<u8> = message.unwrap();
        let xid: XID = XID::new(1, id_generator::gen_id());
        let sender_service_key = new_http_service_key();
        let receiver_service_key = new_receiver_service_key(invoke_request);
        let mut v1_writer = ProtocolV1Writer::new(
            message.len() as u32,
            xid.request_id,
            &sender_service_key,
            &receiver_service_key,
        );
        v1_writer.write_conn_id(1);
        v1_writer.write_msg_body(&message);
        v1_writer.write_msg_type(MsgType::NormalMsg);
        let (sender, receiver) = oneshot::channel::<Vec<u8>>();
        airport.add_request_handler(
            xid,
            Box::new(move |_buffer| {
                let _ = sender.send(_buffer);
            }),
        );
        airport.message_in(1, v1_writer.msg_buffer).await;
        // 等待消息返回
        let resp_message = receiver.await;

        if resp_message.is_err() {
            return Ok(Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .body(Body::empty())
                .unwrap());
        }
        let resp_message = resp_message.unwrap();
        // 获取

        let v1_reader = ProtocolV1Reader::new(&resp_message);

        let msg_body = v1_reader.msg_body();

        let mut is = protobuf::CodedInputStream::from_bytes(msg_body);

        let _tag = is.read_raw_tag_or_eof();

        let _rsp_method_name = is.read_string().unwrap();
        let mut status = Status::default();

        info!("msg_body len = {} is = {}", msg_body.len(), is.pos());

        if is.pos() < msg_body.len() as u64 {
            status.parse_from_input_stream_with_tag_and_len(&mut is);
        }

        let json_value = if is.pos() < msg_body.len() as u64 {
            is.read_raw_tag_or_eof().unwrap();
            let value_size = is.read_uint32().unwrap();

            is.push_limit(value_size as u64).unwrap();

            response_proto_to_json_value(&method, &service.api_descriptor, &mut is)
        } else {
            None
        };

        let resp_message = ResponseMessage {
            err_code: status.err_code,
            err_msg: status.err_msg,
            data: json_value,
        };
        let response = Response::builder()
            .header("Content-Type", "application/json")
            .body(Body::from(serde_json::to_string(&resp_message).unwrap()))
            .unwrap();

        Ok(response)
    }

    pub(super) async fn dispatch_request(req: Request<Body>) -> Result<Response<Body>, Infallible> {
        let req_method = req.method();
        match req.method() {
            &Method::POST | &Method::GET => {
                let path = req.uri().path();
                if "/invoke".eq_ignore_ascii_case(path) {
                    return if req_method == &Method::POST {
                        handle_invoke(req).await
                    } else {
                        Ok(Response::builder()
                            .status(StatusCode::METHOD_NOT_ALLOWED)
                            .body(Body::from("/invoke 只支持 POST 请求"))
                            .unwrap())
                    };
                }
                //
                unsafe {
                    //
                    if let Some(captures) = DOWNLOAD_REGX.assume_init_ref().captures(path) {
                        return if req_method == &Method::GET {
                            let download_key = captures.name("key").unwrap().as_str();
                            let download_key = String::from(download_key);
                            handle_download(req, download_key).await
                        } else {
                            Ok(Response::builder()
                                .status(StatusCode::METHOD_NOT_ALLOWED)
                                .body(Body::from("/download/{key} 只支持 GET 请求"))
                                .unwrap())
                        };
                    }
                    //
                    if let Some(captures) = UPLOAD_REGX.assume_init_ref().captures(path) {
                        return if req_method == &Method::POST {
                            let upload_key = captures.name("key").unwrap().as_str();
                            let upload_key = String::from(upload_key);
                            return handle_upload(req, upload_key).await;
                        } else {
                            Ok(Response::builder()
                                .status(StatusCode::METHOD_NOT_ALLOWED)
                                .body(Body::from("/upload/{key} 只支持 GET 请求"))
                                .unwrap())
                        };
                    }
                }
                warn!("请求方法 {} 不存在", path);
                Ok(Response::builder()
                    .status(StatusCode::NOT_FOUND)
                    .body(Body::from("Method Found"))
                    .unwrap())
            }
            _ => Ok(Response::builder()
                .status(405)
                .body(Body::from("Method Not Allowed"))
                .unwrap()),
        }
    }
}

pub struct HttpNet {
    listen_port: u16,
    listen_ip: String,
    is_close: AtomicBool,
    is_exit: AtomicBool,

    http_clent_map: Mutex<HashMap<i64, Arc<tokio::sync::Mutex<Box<HttpClient>>>>>,
}

impl HttpNet {
    pub fn new() -> Self {
        HttpNet {
            listen_ip: String::default(),
            listen_port: 0,
            http_clent_map: Mutex::new(HashMap::new()),
            is_close: AtomicBool::new(false),
            is_exit: AtomicBool::new(false),
        }
    }
    /**
     * 开始
     */
    pub async fn start(self: Arc<Box<HttpNet>>, statrup_noticer: Sender<bool>) {
        let listen_ip_and_port = format!("{}:{}", self.listen_ip, self.listen_port);
        let listener = TcpListener::bind(listen_ip_and_port).await;
        if listener.is_err() {
            error!(
                "启动 http 服务: {}:{} 失败！",
                self.listen_ip, self.listen_port
            );
            statrup_noticer.send(false).unwrap();
            return;
        }
        let listener = listener.unwrap();

        info!("启动 http 服务: {}:{}", self.listen_ip, self.listen_port);

        statrup_noticer.send(true).unwrap();

        router::init();
        //
        while !self.is_close.load(std::sync::atomic::Ordering::Relaxed) {
            let socket = listener.accept().await;
            if socket.is_err() {
                warn!("连接出错...");
                continue;
            }

            let (tcp_stream, _) = socket.unwrap();
            tokio::spawn(async move {
                if let Err(http_err) = Http::new()
                    .http1_only(true)
                    .http1_keep_alive(true)
                    .serve_connection(tcp_stream, service_fn(router::dispatch_request))
                    .await
                {
                    eprintln!("Error while serving HTTP connection: {}", http_err);
                }
            });
        }
        self.is_exit
            .store(true, std::sync::atomic::Ordering::Release);
    }

    pub fn init(&mut self, listen_ip: String, port: u16) {
        self.listen_port = port;
        self.listen_ip = listen_ip;
    }

    /**
     *
     */
    pub fn new_client(&self, request_id: i64) -> Arc<tokio::sync::Mutex<Box<HttpClient>>> {
        let http_client = Box::new(HttpClient::new(request_id));

        let mut http_clent_map = self.http_clent_map.lock().unwrap();

        let http_client = Arc::new(tokio::sync::Mutex::new(http_client));

        http_clent_map.insert(request_id, http_client.clone());

        http_client
    }

    pub fn get_http_client(
        &self,
        request_id: i64,
    ) -> Option<Arc<tokio::sync::Mutex<Box<HttpClient>>>> {
        let http_clent_map = self.http_clent_map.lock().unwrap();

        let http_client = http_clent_map.get(&request_id);

        if http_client.is_none() {
            None
        } else {
            Some(http_client.unwrap().clone())
        }
    }

    pub fn remove_http_client(&self, request_id: i64) {
        //
        let mut http_clent_map = self.http_clent_map.lock().unwrap();

        http_clent_map.remove(&request_id);
    }

    /**
     * 关闭
     */
    pub async fn stop(&self) {
        self.is_close
            .store(true, std::sync::atomic::Ordering::Relaxed);

        while self.is_exit.load(std::sync::atomic::Ordering::Relaxed) {
            sleep(Duration::from_millis(500)).await;
        }
    }
}
